import { Readable, Stream, Transform } from 'stream';

export async function convertStreamToBuffer(stream: Stream): Promise<Buffer> {
  return new Promise<Buffer>((resolve, reject) => {
    const buffers: Buffer[] = [];
    stream.on('data', buffer => buffers.push(typeof buffer === 'string' ? Buffer.from(buffer) : buffer));
    stream.on('end', () => resolve(Buffer.concat(buffers)));
    stream.on('error', error => reject(error));
    (<any>stream).resume();
  });
}

type BufferEncoding = 'ascii' | 'utf8' | 'utf-8' | 'utf16le' | 'ucs2' | 'ucs-2' | 'base64' | 'latin1' | 'binary' | 'hex';

export async function convertStreamToString(stream: Stream, encoding: BufferEncoding = 'utf-8'): Promise<string> {
  const buffer = await convertStreamToBuffer(stream);
  return buffer.toString(encoding);
}

export function convertBufferToStream(buffer: Buffer): Stream {
  return new Readable({
    read() {
      this.push(buffer);
      this.push(null);
    }
  });
}

export class ChunkSplitTransform extends Transform {
  private _buffer;
  private _separator;
  
  constructor(options) {
    super(options);
    this._buffer = '';
    this._separator = options.separator || '\n';
  }
  
  _transform(chunk, encoding, done) {
    let sepPos;
    this._buffer += chunk.toString();
  
    while ((sepPos = this._buffer.indexOf(this._separator)) !== -1) {
      const portion = this._buffer.substr(0, sepPos);
      this.push(portion);
      this._buffer = this._buffer.substr(sepPos + this._separator.length);
    }
  
    done();
  }
  
  _flush(done) {
    this.push(this._buffer);
    done();
  }
}